/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.RedisException;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.RedisCommand.ValueType;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.client.protocol.convertor.IntegerReplayConvertor;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.LockPubSub;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.util.Timeout;
import io.netty.util.TimerTask;

/**
 * 可重入非公平锁。
 * 每当获取一把锁，都会新创建一个RedissonLock实例，该实例的存活时间和创建该实例的线程的存活时间相同
 * <p>
 * Distributed implementation of {@link java.util.concurrent.locks.Lock}
 * Implements reentrant lock.<br>
 * Lock will be removed automatically if client disconnects.
 * <p>
 * Implements a <b>non-fair</b> locking so doesn't guarantees an acquire order.
 *
 * @author Nikita Koksharov
 */
public class RedissonLock extends RedissonExpirable implements RLock {

    /**
     * static内部类。所有被看门狗自动续期管理的锁会以 ExpirationEntry 抽象的形式保存，并放到Map中.
     * 看门狗Map中Entry的键是ConnectionId:lockName。只能区分不同客户端链接，不能区分线程.
     * Entry内部再用threadIds区分线程。但如果查看所有构建 ExpirationEntry 的代码就会发现:
     * 一个ExpirationEntry中只会保存唯一一个ThreadId。
     */
    public static class ExpirationEntry {
        // ThreadId-重入次数
        private final Map<Long, Integer> threadIds = new LinkedHashMap<>();
        private volatile Timeout timeout;

        public ExpirationEntry() {
            super();
        }

        public void addThreadId(long threadId) {
            Integer counter = threadIds.get(threadId);
            if (counter == null) {
                counter = 1;
            } else {
                counter++;
            }
            threadIds.put(threadId, counter);
        }

        public boolean hasNoThreads() {
            return threadIds.isEmpty();
        }

        public Long getFirstThreadId() {
            if (threadIds.isEmpty()) {
                return null;
            }
            return threadIds.keySet().iterator().next();
        }

        public void removeThreadId(long threadId) {
            Integer counter = threadIds.get(threadId);
            if (counter == null) {
                return;
            }
            counter--;
            if (counter == 0) {
                threadIds.remove(threadId);
            } else {
                threadIds.put(threadId, counter);
            }
        }


        public void setTimeout(Timeout timeout) {
            this.timeout = timeout;
        }

        public Timeout getTimeout() {
            return timeout;
        }

    }

    private static final Logger log = LoggerFactory.getLogger(RedissonLock.class);

    // 一个static final的Map，内部的Entry就是看门狗管理的自动续期的锁抽象，使用EntryName作为key
    // 注意，static声明该Map是静态变量，由RedissonLock元类管理，与实例无关，因此所有的RedissonLock实例中的看门狗都将自动续期锁存放于此
    // final声明EXPIRATION_RENEWAL_MAP的引用地址在当前实例中不可变
    private static final ConcurrentMap<String, ExpirationEntry> EXPIRATION_RENEWAL_MAP = new ConcurrentHashMap<>();

    protected long internalLockLeaseTime;

    // 定义当前RedissonLock实例中的两个不可变常量
    final String id;
    final String entryName;

    // 当前RedissonLock实例中的 LockPubSub 的引用地址不可变(相当于单例Bean)。
    // protected声明该变量对RedissonLock的子类和同包可见。
    protected final LockPubSub pubSub;

    // ConnectionManager管理的commandExecutor。
    final CommandAsyncExecutor commandExecutor;

    /**
     * 当使用redissonClient.getXXXLock("lockName")时，调用Redisson类中对应的方法生成对应的锁实例。
     * 例如可重入非公平锁就是 `return new RedissonLock(connectionManager.getCommandExecutor(), name);` 生成。
     * !!!!注意，由于Redisson类实现了RedissonClient，而RedissonClient会以单例Bean模式自动注入到容器中，且ConnectionManager是一个final变量。
     * 因此容器中所有的connectionManager.getCommandExecutor()都是返回相同引用地址的CommandAsyncExecutor。
     * 再传递到这里的构造方法，那么即便是不同的Lock实例，接收的CommandAsyncExecutor也都是相同引用实例的。
     * <p>
     * 因此，虽然RedissonLock中的pubSub和commandExecutor属性并没有定义static，但是经过上面这套传递过程之后，
     * 不同的RedissonLock实例中，pubSub和commandExecutor其实是相同引用地址的属性!!!
     * 这非常重要!!!
     *
     * @param commandExecutor ConnectionManager管理的commandExecutor
     * @param name            锁的名称。例如lockName
     */
    public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
        this.commandExecutor = commandExecutor;
        // 当前分布式结点与redis链接的ConnectionId。在所有的分布式集群中是唯一的。
        // 但是注意，当前结点可能与redis建立很多链接，这些的ConnectionId都是相同的，只有ThreadId不同
        this.id = commandExecutor.getConnectionManager().getId();
        // Redisson看门狗默认的锁的过期时间
        this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();
        // 由 ConnectionId:lockName 构成的entryName。
        this.entryName = id + ":" + name;
        // 显然所有的RedissonLock实例中的pubSub将是同一个
        this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
    }

    protected String getEntryName() {
        return entryName;
    }

    String getChannelName() {
        return prefixName("redisson_lock__channel", getName());
    }

    /**
     * 锁的名称，是以 ConnectionId:ThreadId 构建的
     * ConnectionId:ThreadId
     */
    protected String getLockName(long threadId) {
        return id + ":" + threadId;
    }

    /**
     * 锁不过期的阻塞lock()
     */
    @Override
    public void lock() {
        try {
            lock(-1, null, false);
        } catch (InterruptedException e) {
            throw new IllegalStateException();
        }
    }

    /**
     * 携带锁的过期时间的阻塞lock()
     */
    @Override
    public void lock(long leaseTime, TimeUnit unit) {
        try {
            lock(leaseTime, unit, false);
        } catch (InterruptedException e) {
            throw new IllegalStateException();
        }
    }

    /**
     * 锁不过期且能够响应中断的阻塞lock()
     *
     * @throws InterruptedException
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        lock(-1, null, true);
    }

    /**
     * 携带过期时间且能够响应中断的阻塞lock()
     *
     * @throws InterruptedException
     */
    @Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        lock(leaseTime, unit, true);
    }

    /**
     * 上面的四个阻塞lock()方法的统一实现方法
     *
     * @param leaseTime     锁的过期时间。如果没有指定则为-1，否则是指定的过期时间
     * @param unit          时间单位
     * @param interruptibly 阻塞是否响应中断
     * @throws InterruptedException
     */
    private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        // 获取当前ThreadId。ConnectionId和ThreadId是redisson保证分布式安全的前提
        long threadId = Thread.currentThread().getId();

        // 转到自身的 tryAcquire() 方法。
        // 如果成功获取到锁ttl=null，如果获取锁失败则ttl为已存在锁的剩余存活时间
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // 在tryAcquire()中还会完成看门狗的任务分配

        // 说明成功获取到锁，直接退出
        if (ttl == null) {
            return;
        }

        // 否则说明获取锁失败，lock()会进入阻塞。Redisson并不是零间隔自旋去redis，而是结合订阅机制，先阻塞等待锁剩余存活时间后再去redis。
        // 期间如果锁被提前释放，订阅机制也能终止等待
        // 当前线程订阅一个与目标锁相关的channel，channel的名字固定为 'redisson_lock__channel:{lockName}'
        // 转到下方的subscribe()方法
        // 订阅的目的就是监听指定锁的释放通知。
        RFuture<RedissonLockEntry> future = subscribe(threadId);

        // 根据阻塞是否响应中断信号做不同的设置
        if (interruptibly) {
            commandExecutor.syncSubscriptionInterrupted(future);
        } else {
            commandExecutor.syncSubscription(future);
        }

        // lock()的阻塞是怎么体现的?就是在这里，只要没有获取到锁就会一直在while中重试
        try {
            while (true) {
                // 每次进入循环都调用tryAcquire尝试获取锁
                ttl = tryAcquire(leaseTime, unit, threadId);
                // 否则就是获取失败，ttl代表锁剩余的存活时间
                if (ttl == null) {
                    break;
                }

                // lock()的阻塞逻辑是这样的:
                // 获取到锁剩余存活时间后，则调用JUC的信号量机制去tryAcquire(ttl)
                // 也就是说并不会再去redis获取锁了，而是用JUC来阻塞(内部仍然是自旋获取锁)
                // 期间如果锁被提前释放，订阅回调机制就会释放信号量，所有JUC阻塞的线程先抢信号量，抢到的线程再进入while尝试获取锁。
                // 这是Redisson减少自旋 + 减少无效访问redis次数的方法
                // 如果期间锁没有被提前释放，tryAcquire(ttl)会JUC阻塞ttl后，再次进入while尝试获取锁，于是又是相同的逻辑。
                if (ttl >= 0) {
                    try {
                        // 上面的future分析中说了，future总信号量就是1，所以只要有线程提前拿到了锁，那么这里的tryAcquire将始终阻塞ttl毫秒，除非被锁释放的通知提前唤醒。
                        // 通知的代码在 PublishSubscribe的onMessage()中，其实就是立刻给通道的latch释放一个信号量。
                        // 于是所有订阅同一个通道的且被tryAcquire()阻塞的线程就立刻去争抢这个一个信号量，谁抢到了，谁就能从这里的try退出。抢不到就接着在这try
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
                    if (interruptibly) {
                        future.getNow().getLatch().acquire();
                    } else {
                        future.getNow().getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
            // 如果获取到所或者break退出，都取消订阅
            unsubscribe(future, threadId);
        }
//        get(lockAsync(leaseTime, unit));
    }

    /**
     * tryAcquire方法，尝试获取锁。通过返回值显示成功与否
     *
     * @param leaseTime 如果获取锁成功，则锁的过期时间。-1为不过期
     * @param unit      时间单位
     * @param threadId  发起Acquire的线程的Id
     * @return 如果成功Acquire则返回Null，>=0时代表获取锁失败时已存在锁的剩余存活时间
     */
    private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
        // 调用自身的异步tryAcquire()方法，转到下方的tryAcquireAsync()方法
        // tryAcquireAsync()返回一个RFuture<Long>异步句柄，get(RFuture)等同于CompletableFuture.get()，都是强制等待异步操作的结果
        return get(tryAcquireAsync(leaseTime, unit, threadId));
    }

    /**
     * 立刻返回的tryLock
     *
     * @param leaseTime
     * @param unit
     * @param threadId
     * @return Boolean显示是否成功。使用EVAL_NULL_BOOLEAN将LUA脚本返回的Null(加锁成功)转换为true，将LUA脚本返回的非Null(已存在锁的剩余ttl)转换为false
     */
    private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, TimeUnit unit, long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        }
        RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            if (e != null) {
                return;
            }
            // lock acquired
            if (ttlRemaining) {
                scheduleExpirationRenewal(threadId);
            }
        });
        return ttlRemainingFuture;
    }

    /**
     * 异步tryAcquire方法。
     *
     * @param <T>       将被设置为Long
     * @param leaseTime 如果获取锁成功，则锁的过期时间。-1为不过期
     * @param unit      时间单位
     * @param threadId  发起Acquire的线程的Id
     * @return RFuture<Long>句柄。携带的Long为null时代表锁获取成功，>=0时代表获取失败时已存在锁的剩余存活时间
     */
    private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
        // 如果指定了锁的过期时间，则将其作为参数调用自身的 tryLockInnerAsync()来获取锁并直接返回结果
        if (leaseTime != -1) {
            // 转到下方  tryLockInnerAsync()方法
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }

        // 如果没有指定锁的过期时间，就有可能出现死锁的情况。redisson的解决办法是引入看门狗机制。
        // 看门狗将给锁设置一个过期时间(默认30s)，每隔过期时间的1/3，就会检查该锁是否仍然被创建锁的线程所持有，如果是则重新设置锁的过期时间为30s
        // 将Redisson看门狗的默认过期时间传入 tryLockInnerAsync()方法，此处和普通锁是一致的
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(
                commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                TimeUnit.MILLISECONDS,
                threadId,
                RedisCommands.EVAL_LONG);

        // 普通会 RFuture<Long>.get() 来等待获取锁操作的完成
        // 看门狗接管锁则会 RFuture<Long>.onComplete() 在等待获取锁完成后进一步添加任务

        // 使用看门狗接管这个锁的自动续期。为看门狗添加该锁自动续期的任务
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            // 当ttlRemainingFuture完成时执行该回调方法。ttlRemaining就是ttlRemainingFuture的返回值Long，e是error，这里和Future编排完全一样
            if (e != null) {
                return;
            }
            // 当ttlRemaining为null时说明获取到了锁，则使用看门狗接管该不过期锁的自动续期。
            if (ttlRemaining == null) {
                // 调用下方 scheduleExpirationRenewal 方法。
                scheduleExpirationRenewal(threadId);
            }
        });

        return ttlRemainingFuture;
    }

    /**
     * 只尝试一次的非阻塞加锁。无论是否成功均立刻返回
     *
     * @return
     */
    @Override
    public boolean tryLock() {
        return get(tryLockAsync());
    }

    /**
     * 看门狗程序本体!!!
     * 使用定时任务来为锁进行自动续期。
     * 从本方法的代码中也可以看出，每个锁自带一个看门狗。这个看门狗只会管所在线程上的锁，压根不管别人的，因为没必要也管不着。
     * 线程A获取锁A，自带看门狗A来为锁A执行自动续期。看门狗A的生命周期和锁A保持相同，更和线程A保持相同。
     * 因为如果线程A突然中断，看门狗A会随着一起中断，那么锁A就不会再自动续期了，就会过期，这样就避免了死锁。
     */
    private void renewExpiration() {
        // 从final的看门狗 EXPIRATION_RENEWAL_MAP 中提取出锁 ConnectionId:lockName 对应的锁抽象
        // 如果ee存在，只能代表ee是当前线程所在的服务节点发送给redis的，并不能确定这个锁就是被当前线程持有
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());

        if (ee == null) {
            return;
        }

        // 定义一个定时任务来执行自动续期。定时周期就是 internalLockLeaseTime / 3， 而internalLockLeaseTime其实就是看门狗的默认过期时间30秒
        // 注意Timeout是Netty的库类，说明Redisson底层也是在用Netty来做通信
        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                // 从final的看门狗 EXPIRATION_RENEWAL_MAP 中提取锁 ConnectionId:lockName 对应的锁抽象
                // 如果ee存在，只能代表ee是当前线程所在的服务节点发送给redis的，并不能确定这个锁就是被当前线程持有
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }

                // 从ee中获取ThreadId，即查看这个ee到底是不是被当前线程所持有
                // 但实际上redisson根本不会对ThreadId做判断。这是因为如果查看所有ExpirationEntry的new处就能发现:
                // ExpirationEntry中的ThreadIdMap最多保存一个ThreadId的key，即持有该锁的ThreadId，也就是当前线程Id
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }

                // 调用下方 renewExpirationAsync() 执行自动续期
                RFuture<Boolean> future = renewExpirationAsync(threadId);

                // 回调函数
                future.onComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock " + getName() + " expiration", e);
                        return;
                    }
                    if (res) {
                        // reschedule itself
                        renewExpiration();
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        ee.setTimeout(task);
    }

    /**
     * 将没有指定过期时间的锁纳入到看门狗自动续期的管理中。
     * 其实就是将锁抽象成 ExpirationEntry 并添加到全局的 EXPIRATION_RENEWAL_MAP 中统一管理。
     *
     * @param threadId 发起获取锁线程的id
     */
    private void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        // EXPIRATION_RENEWAL_MAP 会使用 ConnectionId:lockName 保存锁
        // 由于存在重入锁的情况，因此先看看是否已经有之前的重入看门狗
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        // 但是只有ConnectionId和ThreadId同时使用才能唯一标定分布式线程。因此还需要保存ThreadId
        // 总之就是向抽象 ExpirationEntry 中保存当前锁的唯一标识
        if (oldEntry != null) {
            // 对于重入的看门狗目标，不需要额外操作。因为第一次进入时已经设置了监听
            oldEntry.addThreadId(threadId);
        } else {
            entry.addThreadId(threadId);
            // 对于第一次进入的看门狗目标，为其关联看门狗任务
            renewExpiration();
        }
    }

    /**
     * 直接就是将看门狗监听的ThreadId发给redis，原子更新该线程持有的锁。
     * 由于没有做前置过滤，所以可能存在该ThreadId已经不持有锁的情况
     *
     * @param threadId 当前线程看门狗监听的锁
     * @return
     */
    protected RFuture<Boolean> renewExpirationAsync(long threadId) {
        // KEYS[1] = getName() = 锁名称lockName
        // ARGV[1] = internalLockLeaseTime = 看门狗默认过期时间
        // ARGV[2] = getLockName(threadId) = ConnectionId:ThreadId
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // 将当前线程持有的锁的过期时间续期
                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        // 成功返回1，失败返回0
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Collections.<Object>singletonList(getName()),
                internalLockLeaseTime, getLockName(threadId));
    }

    /**
     * 取消看门狗对锁的自动续期管理
     *
     * @param threadId
     */
    void cancelExpirationRenewal(Long threadId) {
        // 从final的看门狗Map中提取出当前线程所在的 ConnectionId:lockName 对应的锁抽象
        // 如果task存在，只能代表task是当前线程所在的服务节点发送给redis的，并不能确定这个锁就是被当前线程持有
        ExpirationEntry task = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (task == null) {
            return;
        }
        // 同样，redisson压根没有去判断当前线程Id是否等于task的ThreadId。反正task也只有一个ThreadId，直接丢进去是了
        if (threadId != null) {
            // 尝试从task中删除当前线程Id(当然可能task已不被当前线程持有，那就啥也不删就是了)
            task.removeThreadId(threadId);
        }

        // 如果task已经不被任何ThreadId持有，就删除其定时任务即可。
        if (threadId == null || task.hasNoThreads()) {
            Timeout timeout = task.getTimeout();
            if (timeout != null) {
                timeout.cancel();
            }
            EXPIRATION_RENEWAL_MAP.remove(getEntryName());
        }
    }

    /**
     * 现在需要先说明一下Redisson在Redis中的存储结构，即如何解决分布式安全。
     * - 在redis中保存一个哈希表。哈希表的名称就是目标锁的名称，假设就叫lockName。这个哈希表本身将作为锁，内部的记录用于解决分布式安全。
     * - 这个哈希表中保存的k-v对，key使用`ConnectionId:ThreadId`来对链接线程进行唯一标识，value是这个链接线程对当前锁的重入次数。
     */

    /**
     * 真正与Redis建立通信，将一系列加锁操作以LUA脚本发送给Redis，确保操作的原子性。
     *
     * @param leaseTime 锁的过期时间。要么是用户指定的过期时间，要么是看门狗设置的过期时间(前序步骤已经将永远存活的锁的-1更改为看门狗过期时间)
     * @param unit      时间单位
     * @param threadId  发起获取锁的线程Id
     * @param command   EVAL_LONG:将LUA脚本返回的值原封返回。EVAL_NULL_BOOLEAN:LUA脚本返回null则转换为返回true，LUA脚本返回非null则转换为返回false
     * @param <T>       同上。可能是Long可能是Boolean
     * @return RFuture<T>
     */
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        // 制作LUA脚本发送给Redis，返回redis原子执行后的结果
        // LUA脚本中的参数:
        // KEYS[1] = getName() = 锁的名称 lockName
        // ARGV[1] = internalLockLeaseTime = 锁过期时间的毫秒格式
        // ARGV[2] = getLockName(threadId) = ConnectionId:ThreadId，例如 11bb52bc-a764-4649-8b46-a61513d7fe44:2304
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                // 如果redis中不存在 lockName 这个key
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        // 创建一个名为 lockName 的哈希表，并向其中添加一条 'key为11bb52bc-a764-4649-8b46-a61513d7fe44:2304, value为1' 的记录
                        "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                        // 为 lockName 的哈希表设置过期时间
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        // 代表成功创建锁，返回Nil
                        "return nil; " +
                        "end; " +

                        // 如果redis中存在 lockName 哈希表，且表中存在 11bb52bc-a764-4649-8b46-a61513d7fe44:2304这个key
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        // 代表是同一个链接线程重入。将 11bb52bc-a764-4649-8b46-a61513d7fe44:2304对应的value值原子自增
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        // 重入之后重新设置过期时间
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        // 返回Nil
                        "return nil; " +
                        "end; " +

                        // 两个if都没进入，代表获取锁失败，返回锁的剩余存活时间。
                        "return redis.call('pttl', KEYS[1]);",
                Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

    private void acquireFailed(long threadId) {
        get(acquireFailedAsync(threadId));
    }

    protected RFuture<Void> acquireFailedAsync(long threadId) {
        return RedissonPromise.newSucceededFuture(null);
    }

    /**
     * tryLock()，在超过waitTime后扔未获取到锁则返回，在waitTime内的逻辑与阻塞lock()相同。
     *
     * @param waitTime  the maximum time to aquire the lock
     * @param leaseTime lease time
     * @param unit      time unit
     * @return 成功与否标志
     * @throws InterruptedException
     */
    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        //
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }

        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(threadId);
            return false;
        }

        current = System.currentTimeMillis();
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(threadId);
            return false;
        }

        try {
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(threadId);
                return false;
            }

            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

    /**
     * 当前线程在获取锁失败进入阻塞后，订阅这个锁对应的通道。
     * 方法仅对子类和同包可见.
     * 子类锁可以覆写该方法，通常是改变订阅的通道名称.
     *
     * @param threadId 尝试获取锁的线程Id
     * @return RFuture<RedissonLockEntry>异步句柄。RedissonLockEntry详见元类
     */
    protected RFuture<RedissonLockEntry> subscribe(long threadId) {
        // 调用LockPubSub.subscribe，注意，在最开始就说了，不同的锁实例调用的pubSub实际上是同一个pubSub(其引用地址相同)
        // LockPubSub没有实现subscribe方法，调用其父类 PublishSubscribe.subscribe() 实现。
        // 在重入非公平锁的情况下，两个输入参数分别是 ConnectionId:lockName 和 redisson_lock__channel:{lockName}
        return pubSub.subscribe(getEntryName(), getChannelName());
    }

    protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {
        pubSub.unsubscribe(future.getNow(), getEntryName(), getChannelName());
    }

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        return tryLock(waitTime, -1, unit);
    }

    /**
     * 解锁
     */
    @Override
    public void unlock() {
        try {
            // 调用this.unlockAsync解锁
            get(unlockAsync(Thread.currentThread().getId()));
        } catch (RedisException e) {
            if (e.getCause() instanceof IllegalMonitorStateException) {
                throw (IllegalMonitorStateException) e.getCause();
            } else {
                throw e;
            }
        }

//        Future<Void> future = unlockAsync();
//        future.awaitUninterruptibly();
//        if (future.isSuccess()) {
//            return;
//        }
//        if (future.cause() instanceof IllegalMonitorStateException) {
//            throw (IllegalMonitorStateException)future.cause();
//        }
//        throw commandExecutor.convertException(future);
    }

    @Override
    public Condition newCondition() {
        // TODO implement
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean forceUnlock() {
        return get(forceUnlockAsync());
    }

    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('del', KEYS[1]) == 1) then "
                        + "redis.call('publish', KEYS[2], ARGV[1]); "
                        + "return 1 "
                        + "else "
                        + "return 0 "
                        + "end",
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE);
    }

    @Override
    public boolean isLocked() {
        return isExists();
    }

    @Override
    public RFuture<Boolean> isLockedAsync() {
        return isExistsAsync();
    }

    @Override
    public RFuture<Boolean> isExistsAsync() {
        return commandExecutor.writeAsync(getName(), codec, RedisCommands.EXISTS, getName());
    }

    @Override
    public boolean isHeldByCurrentThread() {
        return isHeldByThread(Thread.currentThread().getId());
    }

    @Override
    public boolean isHeldByThread(long threadId) {
        RFuture<Boolean> future = commandExecutor.writeAsync(getName(), LongCodec.INSTANCE, RedisCommands.HEXISTS, getName(), getLockName(threadId));
        return get(future);
    }

    private static final RedisCommand<Integer> HGET = new RedisCommand<Integer>("HGET", ValueType.MAP_VALUE, new IntegerReplayConvertor(0));

    public RFuture<Integer> getHoldCountAsync() {
        return commandExecutor.writeAsync(getName(), LongCodec.INSTANCE, HGET, getName(), getLockName(Thread.currentThread().getId()));
    }

    @Override
    public int getHoldCount() {
        return get(getHoldCountAsync());
    }

    @Override
    public RFuture<Boolean> deleteAsync() {
        return forceUnlockAsync();
    }

    @Override
    public RFuture<Void> unlockAsync() {
        long threadId = Thread.currentThread().getId();
        return unlockAsync(threadId);
    }

    /**
     * 异步解锁。向redis发送LUA脚本进行解锁。
     * redisson可以确保不会发生跳锁
     *
     * @param threadId ThreadId
     * @return 仍然持有重入锁返回0，完全释放锁返回1，其他情况返回null
     */
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        // KEYS[1] = getName() = 锁的名称 lockName
        // KEYS[2] = getChannelName() = redisson_lock__channel:{lockName}
        // ARGV[1] = LockPubSub.UNLOCK_MESSAGE = 0L
        // ARGV[2] = internalLockLeaseTime = 锁过期时间的毫秒格式
        // ARGV[3] = getLockName(threadId) = ConnectionId:ThreadId，例如 11bb52bc-a764-4649-8b46-a61513d7fe44:2304
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // 如果lockName这个哈希表中不存在11bb52bc-a764-4649-8b46-a61513d7fe44:2304这个key
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        // 说明当前线程以不持有该锁。直接返回nil
                        "return nil;" +
                        "end; " +

                        // 否则将lockName哈希表中的11bb52bc-a764-4649-8b46-a61513d7fe44:2304对应的value原子自减
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        // 如果counter > 0 说明当前线程仍然持有重入锁
                        "if (counter > 0) then " +
                        // 则重新设置锁的过期时间
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        // 并返回0(false)
                        "return 0; " +
                        "else " +
                        // 如果counter = 0 说明当前线程已经完全释放该锁。则删除lockName这个锁
                        "redis.call('del', KEYS[1]); " +
                        // 同时在 redisson_lock__channel:{lockName} 通道上发布锁释放消息 0L
                        "redis.call('publish', KEYS[2], ARGV[1]); " +
                        // 返回1(true)
                        "return 1; " +
                        "end; " +

                        "return nil;",
                Arrays.<Object>asList(getName(), getChannelName()),
                LockPubSub.UNLOCK_MESSAGE,
                internalLockLeaseTime,
                getLockName(threadId));

    }

    /**
     * 异步解锁
     *
     * @param threadId id of thread
     * @return RFuture<Void>. 解锁没有成功与否
     */
    @Override
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        // 调用 this.unlockInnerAsync()解锁。
        // 仍然持有重入锁返回0，完全释放锁返回1，其他情况返回null
        RFuture<Boolean> future = unlockInnerAsync(threadId);

        /**
         * 为异步解锁设置完成回调
         * opStatus: 操作成功与否 Boolean
         * e: error
         */
        future.onComplete((opStatus, e) -> {
            if (e != null) {
                cancelExpirationRenewal(threadId);
                result.tryFailure(e);
                return;
            }

            // null的情况
            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                result.tryFailure(cause);
                return;
            }

            // 取消看门狗对该锁的自动续期管理
            cancelExpirationRenewal(threadId);
            result.trySuccess(null);
        });

        return result;
    }

    @Override
    public RFuture<Void> lockAsync() {
        return lockAsync(-1, null);
    }

    @Override
    public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit) {
        long currentThreadId = Thread.currentThread().getId();
        return lockAsync(leaseTime, unit, currentThreadId);
    }

    @Override
    public RFuture<Void> lockAsync(long currentThreadId) {
        return lockAsync(-1, null, currentThreadId);
    }

    @Override
    public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit, long currentThreadId) {
        RPromise<Void> result = new RedissonPromise<Void>();
        RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId);
        ttlFuture.onComplete((ttl, e) -> {
            if (e != null) {
                result.tryFailure(e);
                return;
            }

            // lock acquired
            if (ttl == null) {
                if (!result.trySuccess(null)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            RFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId);
            subscribeFuture.onComplete((res, ex) -> {
                if (ex != null) {
                    result.tryFailure(ex);
                    return;
                }

                lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
            });
        });

        return result;
    }

    private void lockAsync(long leaseTime, TimeUnit unit,
                           RFuture<RedissonLockEntry> subscribeFuture, RPromise<Void> result, long currentThreadId) {
        RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId);
        ttlFuture.onComplete((ttl, e) -> {
            if (e != null) {
                unsubscribe(subscribeFuture, currentThreadId);
                result.tryFailure(e);
                return;
            }

            // lock acquired
            if (ttl == null) {
                unsubscribe(subscribeFuture, currentThreadId);
                if (!result.trySuccess(null)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            RedissonLockEntry entry = subscribeFuture.getNow();
            if (entry.getLatch().tryAcquire()) {
                lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
            } else {
                // waiting for message
                AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>();
                Runnable listener = () -> {
                    if (futureRef.get() != null) {
                        futureRef.get().cancel();
                    }
                    lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
                };

                entry.addListener(listener);

                if (ttl >= 0) {
                    Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                        @Override
                        public void run(Timeout timeout) throws Exception {
                            if (entry.removeListener(listener)) {
                                lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
                            }
                        }
                    }, ttl, TimeUnit.MILLISECONDS);
                    futureRef.set(scheduledFuture);
                }
            }
        });
    }

    @Override
    public RFuture<Boolean> tryLockAsync() {
        return tryLockAsync(Thread.currentThread().getId());
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long threadId) {
        return tryAcquireOnceAsync(-1, null, threadId);
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, TimeUnit unit) {
        return tryLockAsync(waitTime, -1, unit);
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit) {
        long currentThreadId = Thread.currentThread().getId();
        return tryLockAsync(waitTime, leaseTime, unit, currentThreadId);
    }

    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit,
                                         long currentThreadId) {
        RPromise<Boolean> result = new RedissonPromise<Boolean>();

        AtomicLong time = new AtomicLong(unit.toMillis(waitTime));
        long currentTime = System.currentTimeMillis();
        RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId);
        ttlFuture.onComplete((ttl, e) -> {
            if (e != null) {
                result.tryFailure(e);
                return;
            }

            // lock acquired
            if (ttl == null) {
                if (!result.trySuccess(true)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            long el = System.currentTimeMillis() - currentTime;
            time.addAndGet(-el);

            if (time.get() <= 0) {
                trySuccessFalse(currentThreadId, result);
                return;
            }

            long current = System.currentTimeMillis();
            AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>();
            RFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId);
            subscribeFuture.onComplete((r, ex) -> {
                if (ex != null) {
                    result.tryFailure(ex);
                    return;
                }

                if (futureRef.get() != null) {
                    futureRef.get().cancel();
                }

                long elapsed = System.currentTimeMillis() - current;
                time.addAndGet(-elapsed);

                tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
            });
            if (!subscribeFuture.isDone()) {
                Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                    @Override
                    public void run(Timeout timeout) throws Exception {
                        if (!subscribeFuture.isDone()) {
                            subscribeFuture.cancel(false);
                            trySuccessFalse(currentThreadId, result);
                        }
                    }
                }, time.get(), TimeUnit.MILLISECONDS);
                futureRef.set(scheduledFuture);
            }
        });


        return result;
    }

    private void trySuccessFalse(long currentThreadId, RPromise<Boolean> result) {
        acquireFailedAsync(currentThreadId).onComplete((res, e) -> {
            if (e == null) {
                result.trySuccess(false);
            } else {
                result.tryFailure(e);
            }
        });
    }

    private void tryLockAsync(AtomicLong time, long leaseTime, TimeUnit unit,
                              RFuture<RedissonLockEntry> subscribeFuture, RPromise<Boolean> result, long currentThreadId) {
        if (result.isDone()) {
            unsubscribe(subscribeFuture, currentThreadId);
            return;
        }

        if (time.get() <= 0) {
            unsubscribe(subscribeFuture, currentThreadId);
            trySuccessFalse(currentThreadId, result);
            return;
        }

        long curr = System.currentTimeMillis();
        RFuture<Long> ttlFuture = tryAcquireAsync(leaseTime, unit, currentThreadId);
        ttlFuture.onComplete((ttl, e) -> {
            if (e != null) {
                unsubscribe(subscribeFuture, currentThreadId);
                result.tryFailure(e);
                return;
            }

            // lock acquired
            if (ttl == null) {
                unsubscribe(subscribeFuture, currentThreadId);
                if (!result.trySuccess(true)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            long el = System.currentTimeMillis() - curr;
            time.addAndGet(-el);

            if (time.get() <= 0) {
                unsubscribe(subscribeFuture, currentThreadId);
                trySuccessFalse(currentThreadId, result);
                return;
            }

            // waiting for message
            long current = System.currentTimeMillis();
            RedissonLockEntry entry = subscribeFuture.getNow();
            if (entry.getLatch().tryAcquire()) {
                tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
            } else {
                AtomicBoolean executed = new AtomicBoolean();
                AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>();
                Runnable listener = () -> {
                    executed.set(true);
                    if (futureRef.get() != null) {
                        futureRef.get().cancel();
                    }

                    long elapsed = System.currentTimeMillis() - current;
                    time.addAndGet(-elapsed);

                    tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
                };
                entry.addListener(listener);

                long t = time.get();
                if (ttl >= 0 && ttl < time.get()) {
                    t = ttl;
                }
                if (!executed.get()) {
                    Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                        @Override
                        public void run(Timeout timeout) throws Exception {
                            if (entry.removeListener(listener)) {
                                long elapsed = System.currentTimeMillis() - current;
                                time.addAndGet(-elapsed);

                                tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
                            }
                        }
                    }, t, TimeUnit.MILLISECONDS);
                    futureRef.set(scheduledFuture);
                }
            }
        });
    }


}
